package com.kohburg.baseplate.xdelay.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.kohburg.baseplate.xdelay.config.DelayQueueConfig;
import com.kohburg.baseplate.xdelay.constant.XdelayConstant;
import com.kohburg.baseplate.xdelay.entity.XdelayMessage;
import com.kohburg.baseplate.xdelay.factory.RabbitTemplateFactory;
import com.kohburg.baseplate.xdelay.store.FlashStore;
import com.kohburg.baseplate.xdelay.util.HttpUtils;
import com.kohburg.baseplate.xdelay.util.MessageSeqUtil;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * 延时任务生产 rabbitmq实现
 *
 * @author zenos hua
 * @date 2020-06-01
 */
@Service
public class XdelayMessageService {

    protected final Log logger = LogFactory.getLog(this.getClass());

    @Value("${monitor.url}")
    private String monitorUrl;

    @Autowired
    private RabbitTemplateFactory rabbitTemplateFactory;

    /**
     * 添加延时任务
     *
     * @param xdelayMsg 延迟消息体
     */
    public void add(XdelayMessage xdelayMsg) {
        // 获取消息发送模板
        RabbitTemplate rabbitTemplate = rabbitTemplateFactory.getInstace();

        CorrelationData correlationData = new CorrelationData(MessageSeqUtil.createMessageSeq(XdelayConstant.ORIGN));

        // 发送延迟消息
        rabbitTemplate.convertAndSend(DelayQueueConfig.DELAYED_EXCHANGE_XDELAY, DelayQueueConfig.DELAY_ROUTING_KEY_XDELAY, xdelayMsg, message -> {
            message.getMessageProperties().setDelay((int) xdelayMsg.getDelayTime());
            return message;
        }, correlationData);

        // 放入内存
        FlashStore.message.put(correlationData.getId(), xdelayMsg);
    }

    /**
     * 发送失败信号
     *
     * @param type 0: 投递交换机失败 1：投递队列失败 2：消费失败
     * @param describe 描述
     * @param message 消息
     */
    public void failedSignal(String type, String describe, Object message) {
        this.failedSignal(type, describe, "", "", message);
    }

    /**
     * 发送失败信号
     *
     * @param type 0: 投递交换机失败 1：投递队列失败 2：消费失败
     * @param describe 描述
     * @param exchange 交换机
     * @param routing 路由
     * @param message 消息
     */
    public void failedSignal(String type, String describe, String exchange, String routing, Object message) {
        this.failedSignal(type, describe, exchange, routing, "", message);
    }

    /**
     * 发送失败信号
     *
     * @param type 0: 投递交换机失败 1：投递队列失败 2：消费失败
     * @param describe 描述
     * @param consume 消费者
     * @param message 消息
     */
    public void failedSignal(String type, String describe, String consume, Object message) {
        this.failedSignal(type, describe, "", "", consume, message);
    }

    /**
     * 发送失败信号
     *
     * @param type 0: 投递交换机失败 1：投递队列失败 2：消费失败
     * @param describe 描述
     * @param exchange 交换机
     * @param routing 路由
     * @param consume 消费者
     * @param message 消息
     */
    private void failedSignal(String type, String describe, String exchange, String routing, String consume, Object message) {
        Map<String, Object> parameter = new HashMap<String, Object>();

        parameter.put("errortype", type);
        parameter.put("exchange", exchange);
        parameter.put("routing", routing);
        parameter.put("consumer", consume);
        parameter.put("errordescribe", describe);
        parameter.put("errormessage", JSON.toJSONString(message));

        String result = null;
        try {
            result = HttpUtils.postJson(monitorUrl + "/monitor/fail-signal", new JSONObject(parameter).toJSONString());
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("消息失败信息写入失败");
            logger.error(parameter);
        }

        if (!"ok".equals(result)) {
            logger.error("消息失败信息写入失败");
            logger.error(parameter);
        } else {
            logger.info("失败消息已记录");
        }
    }
}